package com.rrd.dw.mr.all;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HBaseUtil;
import com.rrd.dw.utils.MD5Util;

/**
 * Created by liubaoxin on 2018/05/28.
 */
public class ExportALLXinyanBlackDataToDW extends Configured implements Tool {


    public static final String CRT001 = "\001";

    public static final String[] DATA_COLUMNS = {"name", "idCardNo","mobile",
            "bankno", "version","data"  };// assets

    public static final String[] JOB_COLUMNS = { "finished:boolean", "requested:boolean", "successful:boolean",
            "start_time:date", "end_time:date", "system_id", "query", "try_times:int", "type", "message", "code" };// job

    @Override
    public int run(String[] args) throws Exception {
        System.out.println("开始执行");
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader().getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }

        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum", prop.get("hbase.zookeeper.quorum").toString());
        conf.set("hbase.rootdir", prop.get("hbase.rootdir").toString());

        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("hbase.client.scanner.timeout.period", "120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("mapreduce.task.io.sort.factor", "10");
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length != 4) {
            System.err.println("Usage: DataExportTool <out> <start_time> <end_time>");
            return -1;
        }
        Long startTime = HBaseUtil.parseDate(otherArgs[1]).getTime();
        Long endTime = HBaseUtil.parseDate(otherArgs[2]).getTime();
        String model = otherArgs[3];
        job.setJarByClass(ExportAllAliPayDataToDW.class);
        List<Scan> scans = new ArrayList<>();

        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.setCaching(200);
        scan.setMaxVersions(1);
        scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes("xinyan_black_records"));
        if (!model.equals("ALL")) {
            System.out.println("增量运行.............");
            scan.setTimeRange(startTime, endTime + 14400000);
        } else {
            System.out.println("全量运行.............");

        }

        scans.add(scan);

        TableMapReduceUtil.initTableMapperJob(scans, ExportXinyanDataToDWMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
        MultipleOutputs.addNamedOutput(job, "basedata", TextOutputFormat.class, Text.class, Text.class);

        job.setReducerClass(ExportXinyanDataToDWReducer.class);
        job.setNumReduceTasks(5);
        return job.waitForCompletion(true) ? 0 : 1;

    }

    public static void main(String[] args) throws Exception {
        System.out.println("开始运行主函数");
        Configuration conf = HBaseConfiguration.create();

        int res = ToolRunner.run(conf, new ExportALLXinyanBlackDataToDW(), args);
        System.exit(res);
    }

    public static class ExportXinyanDataToDWMapper extends TableMapper<Text, Text> {
        private Text k = new Text();
        private Text v = new Text();
        private String timestampstr=null;
        private Cell dataCell = null;
        private String type = "";
        private String field = "";


        @Override
        protected void map(ImmutableBytesWritable key, Result value, Context context)
                throws IOException, InterruptedException {
            StringBuffer sb = new StringBuffer("");
            Object[] rowkey = HBaseUtil.decodeRowkey(key.get());
            if (rowkey == null) {
                return;
            }
            this.timestampstr = DateUtil.formateDateStr(new Date(Long.parseLong(rowkey[0].toString())));
            String job_id = rowkey[1].toString();
            dataCell = value.getColumnLatestCell(Bytes.toBytes("inf"),Bytes.toBytes("idCardNo"));
            String id_no= dataCell==null?"": HBaseUtil.getValue(CellUtil.cloneValue(dataCell)).toString();
            String id=MD5Util.getMd5(job_id+timestampstr+id_no);
            sb.append(id).append(CRT001).append(job_id).append(CRT001).append(timestampstr).append(CRT001);
            //data基本信息
            for (String str1 : DATA_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("inf"),
                        Bytes.toBytes(field));
                if (dataCell != null) {
                    Object dataObject = HBaseUtil.getValue(type,
                            CellUtil.cloneValue(dataCell));
                    if("data".equals(field)){
                        sb.append(dataObject.toString().replaceAll("\\\"","\""));
                    }else{
                        sb.append(dataObject.toString());
                    }
                }
                sb.append(CRT001);
            }
            //job信息
            for (String str1 : JOB_COLUMNS) {
                String[] fieldArray = str1.split(":");
                if (fieldArray.length > 1) {
                    field = fieldArray[0];
                    type = fieldArray[1];
                } else {
                    field = fieldArray[0];
                    type = "string";
                }
                dataCell = value.getColumnLatestCell(Bytes.toBytes("job"),
                        Bytes.toBytes(field));
                if (dataCell != null) {
                    sb.append(HBaseUtil.getValue(type,
                            CellUtil.cloneValue(dataCell)));
                }
                sb.append(CRT001);
            }
            k.set(id);
            v.set(sb.substring(0,sb.length()-1).replace("\n", " "));
            context.write(k,v);
            sb = null;
        }

    }

    public static class ExportXinyanDataToDWReducer extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private String kstr=null;
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<>(context);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            this.mos.close();
        }

        private NullWritable k = NullWritable.get();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            kstr = key.toString();
            for (Text val : values) {
                this.mos.write("basedata", k, val);
            }

        }
    }
}
